package com.alison.tableapisql.chapter1_tableapiandsql;

import com.alison.tableapisql.chapter1_tableapiandsql.model.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @Author alison
 * @Date 2024/4/8 10:50
 * @Version 1.0
 * @Description
 */
public class E1_TaleTest1_Example {


    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 并行度设置为1
        env.setParallelism(1);
        // 创建Table执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String path = "D:/workspace/lab/learnbigdata/learnflink/flink-datastream/src/main/java/com/alison/datastreamsimple/chapter8_tableapiandsql/sql/E1.txt";
        // read text
        DataStreamSource<String> inputStream = env.readTextFile("file:///" + path);
        // 转换成pojo
        DataStream<SensorReading> datastream = inputStream.map(line -> {
            String[] split = line.split(",");
            return SensorReading.of(split[0], new Long(split[1]), new Double(split[2]));
        });
        // 将流转换为表
        // id, timestamp, temperature
        Table dataTable = tableEnv.<SensorReading>fromDataStream(datastream);
//        Table dataTable = tableEnv.fromDataStream(datastream, $("id"), $("timestamp"), $("temperature"));
//        Table dataTable = tableEnv.fromDataStream(datastream, $("f0").as("sen"));
        tableEnv.createTemporaryView("sensor", dataTable);
        // 调用table API进行转换操作
//        Table resultTable = dataTable.select("id, temperature")
//                .where($("id").isEqual("sensor_1"));
        // 6. 执行SQL
        tableEnv.executeSql("desc sensor").print();
        String sql = "select id, temperature from sensor where id = 'sensor_1'";
        Table resultSqlTable = tableEnv.sqlQuery(sql);

//        tableEnv.toAppendStream(resultTable, Row.class).print("result");
        tableEnv.toAppendStream(resultSqlTable, Row.class).print("sql");

        env.execute();
        /*

input:
sensor_1,1547718199,35.8
sensor_6,1547718201,15.4
sensor_7,1547718202,6.7
sensor_10,1547718205,38.1
sensor_1,1547718207,36.3
sensor_1,1547718209,32.8
sensor_1,1547718212,37.1

result:

result> sensor_1,35.8
sql> sensor_1,35.8
result> sensor_1,36.3
sql> sensor_1,36.3
result> sensor_1,32.8
sql> sensor_1,32.8
result> sensor_1,37.1
sql> sensor_1,37.1
         */
    }


}
